package com.mangix.scheduler;

import com.mangix.core.MangixNode;
import com.mangix.http.MangixHttpClient;
import com.mangix.http.MangixHttpHeader;
import com.mangix.utils.MangixUtils;
import org.eclipse.jetty.client.HttpClient;
import org.eclipse.jetty.client.api.ContentResponse;
import org.eclipse.jetty.client.api.Request;
import org.eclipse.jetty.client.util.PathContentProvider;

import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.util.Map;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;

/**
 * Transfer the objects that are not belongs to this node to the appropriate nodes.
 * If the transfer is success, it will remove the object in local.
 * Created by ZXL on 2014/6/11.
 */
public class MangixTransfer extends Thread {

    private ConcurrentHashMap<Path, Boolean> taskMap;
    private BlockingQueue<Path> taskQueue;
    private MangixNode localNode;
    private int M;

    public MangixTransfer(ConcurrentHashMap<Path, Boolean> taskMap, BlockingQueue<Path> taskQueue, MangixNode localNode) {
        this.taskMap = taskMap;
        this.taskQueue = taskQueue;
        this.localNode = localNode;
        this.M = localNode.getConfig().getM();
        this.setName("Mangix Object Transmitter - " + localNode.toString());
    }


    private void transfer(MangixNode dest, Path file, String key) throws IOException {
        try {
            MangixHttpClient.put("http://" + dest.getHostName() + ":" + dest.getPort() + "/direct/objects/" + key, file, file.getFileName().toString(), localNode.getFiles().getReplicasNum(file));
            System.out.println(localNode.toString() + " transfer: " + file.toString() + " successful!");
            // Delete local file because it too old or it does not belong to this node.
            Files.deleteIfExists(file);
        } catch (Exception e) {
            e.printStackTrace();
            System.out.println(localNode.toString() + " transfer: " + file.toString() + " failed!");
        }
    }

    public void run() {
        Path file;
        try {
            while ((file = taskQueue.take()) != null) {
                try {
                    String key = file.getParent().getFileName().toString();
                    MangixNode coordinator = localNode.findSuccessor(MangixUtils.hash(key, M));
                    if (coordinator != null && !coordinator.equals(localNode)) {
                        Map meta = coordinator.getMeta(key);
                        if (meta == null) {
                            transfer(coordinator, file, key);
                        } else {
                            long localVersion = Long.parseLong(file.getFileName().toString().split("\\.")[0]);
                            long remoteVersion = (Long)meta.get("version");
                            if (localVersion <= remoteVersion) {
                                Files.deleteIfExists(file);
                            } else if (localVersion > remoteVersion) {
                                transfer(coordinator, file, key);
                            }
                        }
                    }
                    taskMap.remove(file);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }


}
